package run.jingjing.websocket.endpoint;

import cn.hutool.extra.spring.SpringUtil;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.RedisSerializer;
import run.jingjing.websocket.model.LWsGroup;
import run.jingjing.websocket.model.LWsSession;
import run.jingjing.websocket.model.LWsTopic;
import run.jingjing.websocket.template.WebSocketTemplate;
import run.jingjing.websocket.model.RunningTopic;
import run.jingjing.websocket.timer.BroadcastOnlineCountTask;
import run.jingjing.websocket.timer.LocalCleanSessionTask;
import run.jingjing.websocket.timer.LocalOnlineCountTask;
import javax.websocket.Session;
import java.io.IOException;
import java.io.Serializable;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.TimeUnit;

/**
 * @author: jingjing
 * @date: 2021/6/24
 * @description: base endpoint
 */
@Data
@Slf4j
public abstract class AbstractWebSocketEndpoint implements WebSocketEndpoint {

    /**
     * some operations template of run.jingjing.websocket
     */
    protected static WebSocketTemplate webSocketTemplate;

    protected static RedisTemplate redisTemplate;

    /**
     * set of active topics
     */
    private static final CopyOnWriteArraySet<RunningTopic> runningTopics = EndpointManager.runningTopics;

    /**
     * global scheduler, tickTime = 1000ms
     */
    protected static final HashedWheelTimer hashedWheelTimer = new HashedWheelTimer(1, TimeUnit.SECONDS);

    static {
        hashedWheelTimer.start();
    }

    /**
     * initialize token checker
     * @return
     */
    protected abstract boolean checkToken(String token);

    protected LWsSession lWsSession;

    /**
     * get userId from token
     *
     * @param token
     * @return
     */
    protected abstract Integer getUserIdFromToken(String token);

    @Override
    public void beforeInitializationHook() {
        if (webSocketTemplate == null) {
            webSocketTemplate = SpringUtil.getBean("webSocketTemplate");
        }
        if (redisTemplate == null) {
            redisTemplate = SpringUtil.getBean("redisTemplate");
        }
    }

    @Override
    public void initEndpoint(Class<? extends LittlebustersEndpointDemo> clazz, String token, String topicName, Session session) {
        // before hook
        beforeInitializationHook();

        // get params
        if (!checkToken(token)) {
            throw new RuntimeException("valid token");
        }
        Integer userId = getUserIdFromToken(token);
        String groupName = clazz.getName();

        // checkout & put session
        LWsGroup lWsGroup = EndpointManager.getGroup(groupName);
        LWsTopic lWsTopic = lWsGroup.get(topicName);
        LWsSession lWsSession = new LWsSession();
        lWsSession.setSession(session);
        lWsSession.setToken(token);
        lWsSession.setUserId(userId);
        lWsTopic.put(userId, lWsSession);
        lWsSession.setTopic(lWsTopic);
        this.lWsSession = lWsSession;

        // run task
        run(lWsGroup, lWsTopic);

        // after hook
        afterInitializationHook();
    }

    /**
     * by redis, subscribe executors that handle with any events
     *
     * @param runningTopic
     * @return
     */
    private void doSubscribe(RunningTopic runningTopic) {
        RedisMessageListenerContainer container = webSocketTemplate.getRedisMessageListenerContainer();
        RedisSerializer valueSerializer = redisTemplate.getValueSerializer();
        MessageListenerAdapter messageListenerAdapter = new MessageListenerAdapter(new MessageListener() {
            @Override
            public void onMessage(Message message, byte[] pattern) {
                byte[] body = message.getBody();
                EventMessage eventMessage = (EventMessage) valueSerializer.deserialize(body);
                PublishEvent event = eventMessage.getEvent();
                switch (event) {
                    case REGISTRY:
                        doRegistry(eventMessage);
                        break;
                    case SEND_EVENT:
                        doSend(eventMessage);
                        break;
                    case BROADCAST:
                        doBroadcast(eventMessage);
                        break;
                    default:
                }
            }
        });
        container.addMessageListener(messageListenerAdapter, runningTopic.channelTopic());
        runningTopic.setMessageListener(messageListenerAdapter);
        runningTopic.setContainer(container);
    }

    /**
     * timetask running with new topic
     */
    protected void run(LWsGroup lWsGroup, LWsTopic lWsTopic) {
        RunningTopic runningTopic = new RunningTopic();
        runningTopic.setlWsGroup(lWsGroup);
        runningTopic.setlWsTopic(lWsTopic);

        // create task if new topic
        if (runningTopics.add(runningTopic)) {

            // count task that count local session number
            runLocalOnlineCountTask(runningTopic);

            // scan task that is for closed session
            runLocalCleanSessionTask(runningTopic);

            // broadcast task that notify to every session in topic
            runBroadcastOnlineCountTask(runningTopic);

            // start subscribe event that is from others node
            doSubscribe(runningTopic);

            // tell others node, a new session logined in local
            webSocketTemplate.publish(runningTopic, PublishEvent.REGISTRY, webSocketTemplate.getAddress(), null, getUserId());
        }
    }

    private void runBroadcastOnlineCountTask(RunningTopic runningTopic) {
        // interval of this task
        Integer intervalSecond = 15;
        BroadcastOnlineCountTask task = new BroadcastOnlineCountTask(runningTopic, webSocketTemplate, hashedWheelTimer, intervalSecond);
        Timeout timeout = hashedWheelTimer.newTimeout(task, intervalSecond, TimeUnit.SECONDS);
        runningTopic.addTimeouts(timeout);
    }

    private void runLocalCleanSessionTask(RunningTopic runningTopic) {
        // interval of this task
        Integer intervalSecond = 15;
        LocalCleanSessionTask localCleanSessionTask = new LocalCleanSessionTask(runningTopic, hashedWheelTimer, intervalSecond);
        Timeout timeout = hashedWheelTimer.newTimeout(localCleanSessionTask, intervalSecond, TimeUnit.SECONDS);
        runningTopic.addTimeouts(timeout);
    }

    private void runLocalOnlineCountTask(RunningTopic runningTopic) {
        // interval of this task
        Integer intervalSecond = 6;
        LocalOnlineCountTask localOnlineCountTask = new LocalOnlineCountTask(runningTopic, webSocketTemplate, redisTemplate, hashedWheelTimer, intervalSecond);
        Timeout timeout = hashedWheelTimer.newTimeout(localOnlineCountTask, intervalSecond, TimeUnit.SECONDS);
        runningTopic.addTimeouts(timeout);
    }

    public void doRegistry(EventMessage eventMessage) {
        String group = eventMessage.getGroup();
        Integer fromUserId = eventMessage.getFromUserId();
        String topic = eventMessage.getTopic();
        String address = eventMessage.getContent();
        // received message someone client registered in others service
        if (!Objects.equals(webSocketTemplate.getAddress(), address)) {
            LWsTopic lWsTopic = EndpointManager.getTopic(group, topic);
            if (lWsTopic != null && lWsTopic.containsKey(fromUserId)) {
                try {
                    // if same user-session in local, close it
                    lWsTopic.get(fromUserId).getSession().close();
                } catch (IOException e) {
                    log.error("session closing occurred error");
                    throw new RuntimeException("session closing occurred error");
                }
                log.info("user：{} topic:{} is already logined in {} this server will close that connected with this user", topic, fromUserId, address);
            }
        }
    }

    public void doSend(EventMessage eventMessage) {
        String group = eventMessage.getGroup();
        String topic = eventMessage.getTopic();
        Integer toUserId = eventMessage.getToUserId();
        LWsTopic lWsTopic = EndpointManager.getTopic(group, topic);
        if (lWsTopic != null && lWsTopic.containsKey(toUserId)) {
            LWsSession lWsSession = lWsTopic.get(toUserId);
            Session session = lWsSession.getSession();
            try {
                session.getBasicRemote().sendText(eventMessage.getContent());
            } catch (IOException e) {
                log.error("user:{} topic:{} send msg occurred error", lWsSession.getUserId(), topic);
                throw new RuntimeException("send msg occurred error");
            }
        }
    }

    public void doBroadcast(EventMessage eventMessage) {
        String group = eventMessage.getGroup();
        String topic = eventMessage.getTopic();
        LWsTopic lWsTopic = EndpointManager.getTopic(group, topic);
        if (lWsTopic != null) {
            try {
                for (LWsSession lWsSession : lWsTopic.values()) {
                    lWsSession.getSession().getBasicRemote().sendText(eventMessage.getContent());
                }
            } catch (IOException e) {
                log.error("topic:{} broadcast msg occurred error", topic);
                throw new RuntimeException("broadcast msg occurred error");
            }
        }
    }

    public void broadcastOnlineCount(Class<? extends WebSocketEndpoint> clazz, Map<String, ConcurrentHashMap<Integer, WebSocketEndpoint>> container, Long interval, TimeUnit timeUnit) {
        TimerTask timerTask = new TimerTask() {
            @Override
            public void run(Timeout timeout) throws Exception {

            }
        };
        hashedWheelTimer.newTimeout(timerTask, interval, timeUnit);
    }

    @Data
    public static class EventMessage implements Serializable {

        private PublishEvent event;

        private String group;

        private String topic;

        private Integer toUserId;

        private Integer fromUserId;

        private String content;
    }

    @AllArgsConstructor
    public enum MsgSign {
        OnlineCount("在线人数");

        public final String desc;
    }

    @AllArgsConstructor
    public enum PublishEvent {
        SEND_EVENT("给用户定向发送消息事件"),
        BROADCAST("广播事件"),
        REGISTRY("注册事件");

        public final String desc;
    }
}
